import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
//-------------------------------------------------------下面面这个就是动态转换部分------------------------------------
public class RDD2DataFrameByReflection 
{
@SuppressWarnings("deprecation")//非常赞的用法,直接无视警告
public static void main(String[] args) 
{
            // TODO Auto-generated method stub
        SparkConf conf=new SparkConf().setAppName("RDD2DATAFRAME").setMaster("local");
        JavaSparkContext sc=new JavaSparkContext(conf);
        SQLContext sqlcontext=new SQLContext(sc); 
        JavaRDD<String> lines=sc.textFile("person.txt");
        JavaRDD<Person> persons=lines.map(new Function<String,Person>()
            {//这个就是匿名函数


private static final long serialVersionUID = 1L;//序列化时为了保持版本的兼容性，即在版本升级时反序列化仍保持对象的唯一性。


@Override
public Person call(String line) throws Exception //调用类的赋值函数(成员函数),也就是动态转换部分.
{

    String[] splited=line.split(",");
    Person p=new Person();
    p.setId(Integer.valueOf(splited[0].trim()));
    p.setName(splited[1].trim());
    p.setAge(Integer.valueOf(splited[2].trim())); 
    return p;
} 
});




    //第一个参数：RDD，第二个参数是JavaBean，Person类
    //第二参数就是封装的JavaBean，JavaBean中封装了Person的元数据信息，
    //通过第二个参数DataFrame也就获得了元数据信息。
    //在底层通过反射的方式获得Person的所有Fields，结合RDD本身，就生成了DataFrame
    Dataset<Row> df=sqlcontext.createDataFrame(persons, Person.class);//这句话就是使用反射的语句,注意第二个参数Person.class
    //这个其实就是利用反射来提供字段信息,专业点的说法,就是利用反射(Person.class)来提供元数据(字段)

//-------------------------------------------------------上面这个就是动态转换部分------------------------------------

df.registerTempTable("persons"); //注册为临时表
Dataset<Row> bigdatas=sqlcontext.sql("select * from persons where age>=6");

System.out.println("----------------------------------下面的是临时注册表输出结果---------------------------------------");
bigdatas.show();
System.out.println("--------------------------------下面的是Dataframe中的运算结果转化为RDD-----------------------------------------");
JavaRDD<Row> bigdataRDD=bigdatas.javaRDD();
/*
List<Row> resultrdd=bigdataRDD.collect();
for (Row r:resultrdd) {
System.out.println(r);
}
*/

JavaRDD<Person> result=bigdataRDD.map(new Function<Row,Person>()
{ 
    private static final long serialVersionUID = 1L;
    @Override
    public Person call(Row row) throws Exception 
    {
        Person p=new Person();
        p.setId(row.getInt(1));
        p.setName(row.getString(2));
        p.setAge(row.getInt(0));
        return p;
    }
}
);

List<Person> personlist=result.collect();
for(Person p:personlist) 
{
    System.out.println("-------------------------------------------------------------");
    System.out.println(p);
    System.out.println("-------------------------------------------------------------");
}

sc.stop();



}


}

// 参考链接:
// https://blog.csdn.net/sundujing/article/details/51438142
//链接中的元数据指的就是字段